package cn.gwm.flink.streaming.demo.ods;

import org.apache.flink.orc.vector.OrcLongColumnVector;
import org.apache.flink.orc.vector.Vectorizer;
import org.apache.hadoop.hive.ql.exec.vector.BytesColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.TimestampColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;

import java.io.IOException;
import java.io.Serializable;
import java.nio.charset.StandardCharsets;

/**
 * @author GW00283474
 */
public class OdsSignalVectorizer extends Vectorizer<OdsSignal> implements Serializable {
    public OdsSignalVectorizer(String schema) {
        super(schema);
    }

    @Override
    public void vectorize(OdsSignal odsSignal, VectorizedRowBatch batch) throws IOException {
        BytesColumnVector deviceIdColVector = (BytesColumnVector) batch.cols[0];
        BytesColumnVector itemTimeColVector = (BytesColumnVector) batch.cols[1];
        BytesColumnVector itemDateColVector = (BytesColumnVector) batch.cols[2];
        BytesColumnVector ingestionTimeColVector = (BytesColumnVector) batch.cols[3];
        BytesColumnVector ingestionDateColVector = (BytesColumnVector) batch.cols[4];
        BytesColumnVector contentColVector = (BytesColumnVector) batch.cols[5];
        int row = batch.size++;
        deviceIdColVector.setVal(row, odsSignal.getDeviceId().getBytes(StandardCharsets.UTF_8));
        itemTimeColVector.setVal(row, odsSignal.getItemTime().getBytes(StandardCharsets.UTF_8));
        itemDateColVector.setVal(row,odsSignal.getItemDate().getBytes(StandardCharsets.UTF_8));
        ingestionTimeColVector.setVal(row,odsSignal.getIngestionTime().getBytes(StandardCharsets.UTF_8));
        ingestionDateColVector.setVal(row,odsSignal.getIngestionDate().getBytes(StandardCharsets.UTF_8));
        contentColVector.setVal(row, odsSignal.getContent().getBytes(StandardCharsets.UTF_8));
    }
}
